package com.xiguaji.tiktok.task.aweme.event;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.resource.ResourceUtil;
import com.google.common.eventbus.AllowConcurrentEvents;
import com.google.common.eventbus.Subscribe;
import com.google.common.util.concurrent.AtomicDouble;
import com.gugee.tiktok.common.model.dal.blogger.AwemeCommentRepository;
import com.gugee.tiktok.common.model.dal.blogger.AwemeRepository;
import com.gugee.tiktok.common.model.es.ESAweme;
import com.gugee.tiktok.common.model.es.ESAwemeUpsert;
import com.gugee.tiktok.common.model.es.Keyword;
import com.gugee.tiktok.common.model.mysql.Aweme;
import com.xiguaji.boot.eventbus.Event;
import com.xiguaji.boot.eventbus.IEventListener;
import com.xiguaji.boot.toolkit.DateTimeUtils;
import com.xiguaji.boot.toolkit.JsonUtil;
import com.xiguaji.tiktok.task.aweme.client.BloggerClient;
import com.xiguaji.tiktok.task.aweme.spider.AwemeSpiderClient;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;

@Slf4j
@Component
public class ESRefreshAwemeEventListener implements IEventListener<ESRefreshAwemeEventListener.ESRefreshAwemeEvent> {
    private final String AWEME_INDEX = "aweme_index_";
    @Autowired
    AwemeRepository awemeRepository;
    @Autowired
    RestHighLevelClient highLevelClient;
    @Autowired
    AwemeSpiderClient awemeSpiderClient;
    @Autowired
    ESAwemeUpsert.ESAwemeMapper esAwemeMapperUpsert;
    @Autowired
    ESAweme.ESAwemeMapper esAwemeMapper;
    @Autowired
    BloggerClient bloggerClient;
    @Autowired
    AwemeCommentRepository awemeCommentRepository;

    private DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMM");

    @Override
    @Subscribe
    @AllowConcurrentEvents
    public void handle(ESRefreshAwemeEvent event) {
        long start = System.currentTimeMillis();
        List<Aweme> awemes = new ArrayList<>();

        if(null == event.awemeId){
            awemes = awemeRepository.findAwemesByoUidAndoCreateTimeGreaterThan(event.bloggerUid, DateTimeUtils.second(LocalDateTime.now().minusDays(90)));

        }else{
            val awemeOptional = awemeRepository.findByoAwemeId(event.awemeId);

            if(awemeOptional.isPresent()){
                saveAwemeToEs(awemeOptional.get());
                if((System.currentTimeMillis()-start)/1000>5)
                    log.info("ESRefreshAwemeEventListener awemeId:{} use time:{}s",awemeOptional.get().getOUid(),(System.currentTimeMillis()-start)/1000);
                return;
            }
        }
        saveAwemesToEs(awemes,event.bloggerUid);
    }

    private void saveAwemesToEs(List<Aweme> awemes,long bloggerUid){
        long start = System.currentTimeMillis();
        if(awemes.size() > 0){
            BulkRequest bulkRequest = new BulkRequest();
            var bloggerMain = bloggerClient.getBloggerMainByOUid(bloggerUid);
            awemes.forEach( a->{
                ESAwemeUpsert esAweme = esAwemeMapperUpsert.mapper(a);
                esAweme.setUniqueId(bloggerMain.getOUniqueId());
                String jsonAweme=JsonUtil.parseJson(esAweme);
                UpdateRequest updateRequest = new UpdateRequest(AWEME_INDEX+getDateStr(esAweme.getCreateTime()), String.valueOf(esAweme.getId()))
                        .doc(jsonAweme,XContentType.JSON)
                        .upsert(jsonAweme,XContentType.JSON);
                bulkRequest.add(updateRequest);
            });
            try {
                val bulk = highLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                if((System.currentTimeMillis()-start)/1000>5)
                    log.info("ESSaveAweme upsert count:{},use time={},bloggerUid={}",awemes.size(),(System.currentTimeMillis()-start)/1000,bloggerUid);
            } catch (IOException e) {
                log.error("ES RefreshBlogger error",e);
            }
        }
    }

    private void saveAwemeToEs(Aweme aweme) {
        var bloggerMain = bloggerClient.getBloggerMainByOUid(aweme.getOUid());
        ESAweme esAweme = esAwemeMapper.mapper(aweme);
        esAweme.setUniqueId(bloggerMain.getOUniqueId());

        List<Keyword> keywordList = getKeyword(esAweme.getAwemeId());

        if(null != keywordList && keywordList.size() > 0){
            esAweme.setAwemeTags(keywordList);
        }
        List<Keyword> keywords = findKeywordByES(aweme);

        if(keywords.size() > 0){
            esAweme.setAwemeTags(keywords);
        }
        IndexRequest indexRequest = new IndexRequest(AWEME_INDEX+getDateStr(esAweme.getCreateTime()));
        indexRequest.id(String.valueOf(esAweme.getId()));
        indexRequest.source(JsonUtil.parseJson(esAweme), XContentType.JSON);

        try {
            val response = highLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("ES RefreshBlogger error",e);
        }
    }

    private List<Keyword> findKeywordByES(Aweme aweme){
        GetRequest getRequest = new GetRequest(AWEME_INDEX+getDateStr(aweme.getOCreateTime()),String.valueOf(aweme.getId()));
        FetchSourceContext fetchSourceContext = new FetchSourceContext(true);
        getRequest.fetchSourceContext(fetchSourceContext);
        GetResponse getResponse;

        try {
            getResponse = highLevelClient.get(getRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error("ESRefreshAwemeEventListener findEs error!",e);
            return new ArrayList<>();
        }



        if(null == getResponse.getSource()){
            return new ArrayList<>();
        }

        if(null == getResponse.getSource().get("awemeTags")){
            return new ArrayList<>();
        }

        return (List<Keyword>) getResponse.getSource().get("awemeTags");
    }

    private String getDateStr(long second){
        return DateUtil.format(DateUtil.calendar(second*1000).getTime(),"yyyyMM");
    }

    private List<Keyword> getKeyword(long awemeId){
        Map<Integer,String> requestBody = getRequestBody(awemeId);

        if(requestBody.isEmpty()){
            return null;
        }

        List<String> commentAnalyzed = null;

        try{

            long start = System.currentTimeMillis();
            commentAnalyzed = awemeSpiderClient.getCommentAnalyzed(requestBody);
            if((System.currentTimeMillis()-start)/1000>10)
                log.info("ESRefreshAwemeEventListener awemeId: "+awemeId+" use time:"+(System.currentTimeMillis()-start)/1000+"s");
        }catch (Exception e){
            log.error("awene ai comment error! awemeId:{} , params:{}", awemeId,JsonUtil.parseJson(requestBody),e);
        }

        if(null == commentAnalyzed || commentAnalyzed.size() < 1 ){
            return null;
        }
        Map<String,Integer> rels = new HashMap<>();
        List<Keyword> keywords = new ArrayList<>();

        AtomicDouble keywordCount = new AtomicDouble();

        for (String l : commentAnalyzed) {
            if(null == rels.get(l)){
                int count = Collections.frequency(commentAnalyzed,l);
                rels.put(l,count);
                keywordCount.addAndGet(count);
            }
        };

        rels.forEach((k,v)->{
            Keyword keyword = new Keyword();
            keyword.setKeyword(k);
            keyword.setPercent(((int)(v/keywordCount.get()*100*100))/100.0);
            keywords.add(keyword);
        });


        if(keywords.size() > 1){
            Collections.sort(keywords,(first,second)-> {
                if(first.getPercent() > second.getPercent()){
                    return -1;
                } else if (first.getPercent() == second.getPercent()){
                    return 0;
                }
                return 1;
            });
        }

        return keywords.size() >10?keywords.subList(0,10):keywords;

    }


    private String getIndex(String index) {

        GetIndexRequest getIndexRequest = new GetIndexRequest(index);
        //getIndexRequest.humanReadable(true);
        try{
            val exists = highLevelClient.indices().exists(getIndexRequest, RequestOptions.DEFAULT);

            if(!exists){
                String mappingJson = ResourceUtil.readUtf8Str("aweme.json");
                CreateIndexRequest createIndexRequest = new CreateIndexRequest(index);
                createIndexRequest.settings(Settings.builder()
                        .put("index.number_of_shards",8)
                        .put("index.number_of_replicas",1));

                createIndexRequest.mapping(mappingJson,XContentType.JSON);
                val createIndexResponse = highLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
                return createIndexResponse.index();
            }
        }catch (Exception e){
            log.error("create index: {} error",index,e);
            return null;
        }



        return index;
    }


    private Map<Integer,String> getRequestBody(long awemeId){
        long start = System.currentTimeMillis();
        Map<Integer,String> body = new HashMap<>();

        try{
            String comment = "";
            val aweme = awemeRepository.findByoAwemeId(awemeId);
            if(aweme.isPresent()){
                comment = awemeCommentRepository.get(awemeId,aweme.get().getDatecode());
            }
            if((System.currentTimeMillis()-start)/1000>5)
                log.info("ESRefreshAwemeEventListener get oss data awemeId:{} use time:{}s",awemeId,(System.currentTimeMillis()-start)/1000);
            List<Map<String,String>> comments = new ArrayList<>();
            comments = JsonUtil.jsonParse(comments.getClass(),comment);

            for (int i = 0; i < comments.size(); i++) {
                body.put(i,comments.get(i).get("text"));
            }
        }catch (Exception e){
            log.error("ESRefreshAwemeEventListener get oss data error!",e.getMessage());
        }

        return body;

    }



    public static class ESRefreshAwemeEvent extends Event {
        public ESRefreshAwemeEvent(Long bloggerUid) {
            super( null,"ESSaveAweme");
            this.bloggerUid = bloggerUid;
        }

        public ESRefreshAwemeEvent(Event event ,Long bloggerUid) {
            super( event,"ESSaveAweme");
            this.bloggerUid = bloggerUid;
        }

        public ESRefreshAwemeEvent(Long bloggerUid,Long awemeId) {
            super( null,"ESSaveAweme");
            this.bloggerUid = bloggerUid;
            this.awemeId = awemeId;
        }

        public ESRefreshAwemeEvent(Event event ,Long bloggerUid,Long awemeId) {
            super( event,"ESSaveAweme");
            this.bloggerUid = bloggerUid;
            this.awemeId = awemeId;
        }

        private Long bloggerUid;

        private Long awemeId;
    }
}
